/*
 * Copyright (c) 2022. China Mobile(SuZhou)Software Technology Co.,Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package meta

import (
	"context"
	"encoding/json"
	"strconv"
	"strings"
	"sync"
	"time"
	"unsafe"

	"github.com/go-redis/redis/v8"
)

const (
	BKPIMAGE_HAS_HOLE      uint8 = 0x01
	BKPIMAGE_IS_COMPRESSED uint8 = 0x02

	BKPBLOCK_FORK_MASK uint8 = 0x0F
	BKPBLOCK_HAS_IMAGE uint8 = 0x10
	BKPBLOCK_SAME_REL  uint8 = 0x80

	XLR_MAX_BLOCK_ID uint8 = 32

	XLR_BLOCK_ID_DATA_SHORT   uint8 = 255
	XLR_BLOCK_ID_DATA_LONG    uint8 = 254
	XLR_BLOCK_ID_ORIGIN       uint8 = 253
	XLR_BLOCK_ID_TOPLEVEL_XID uint8 = 252

	XLP_FIRST_IS_CONTRECORD uint16 = 1
	XLP_LONG_HEADER         uint16 = 2

	BLOCKSZ      int = 8192
	SAMENODE     Ino = 0
	SIZEOFUINT32 int = 4
)

var (
	PreHalf   []byte    
	PreIno    Ino       
	PreLsn    XLogRecPtr = 0
	Linked               = false 
	PreOffset uint64     = 0
	gOffset              = &globalOffset{
		offset: 0,
		ino:    Ino(0),
	}
)

type Oid uint32
type TransactionId uint32
type XLogRecPtr uint64
type RmgrId uint8
type pg_crc32c uint32
type BlockNumber uint32
type RepOriginId uint16
type RelFileNode struct {
	spcNode Oid
	dbNode  Oid
	relNode Oid
}

type XLogPageHeaderShort struct {
	XlpMagic    uint16
	XlpInfo     uint16
	XlpTli      uint32
	XlpPageAddr uint64
	XlpRemLen   uint32
}

type XLogPageHeaderLong struct {
	XLogPageHeaderShort
	XlpSysId      uint64
	XlpSegSize    uint32
	XlpXlogBlcksz uint32
}

type XLogRecord struct {
	Xl_tot_len uint32
	Xl_xid     TransactionId
	Xl_prev    XLogRecPtr
	Xl_end     XLogRecPtr
	Xl_info    uint8
	mtr        bool
	blockNum   uint8
	Xl_rmid    RmgrId
	Xl_crc     pg_crc32c
}

type singleWal struct {
	pref  string
	blkno uint32
	wal   WalLoc
}

type globalOffset struct {
	offset uint64
	ino    Ino
}

var wg sync.WaitGroup

var walLengthChan = make(chan uint32, 1)
var isSpan = false

func DecodeXlogData(ctx Context, meta Meta, ino Ino, data []byte, offset uint64,fileName string) {
	var quit bool = false
	
	nameList := strings.Split(fileName,"/")
	listLen := len(nameList)
	if listLen < 3 || nameList[listLen -2] != WalName || (len(nameList[listLen - 1]) != 24 && !strings.Contains(nameList[listLen - 1],TmpXlog)){
		return
	}

	if gOffset.ino != ino {
		gOffset.ino = ino
		gOffset.offset = offset
	}

	var i = 0
	var subData []byte
	totalLenth := len(data)

	for i <= totalLenth/BLOCKSZ {
		if i < totalLenth/BLOCKSZ {
			subData = data[i*BLOCKSZ : (i+1)*BLOCKSZ]
		} else if i == totalLenth/BLOCKSZ && totalLenth%BLOCKSZ > 0 {
			subData = data[i*BLOCKSZ : i*BLOCKSZ+totalLenth%BLOCKSZ]
		} else {
			break
		}
		DecodeXlogPage(ctx, /*meta,*/ ino, subData, offset+uint64(i*BLOCKSZ), &quit)
		if quit {
			break
		}
		i += 1
	}
}

func DecodeXlogPage(ctx Context, /*meta Meta,*/ ino Ino, data []byte, offset uint64, quit *bool) {

	var (
		totalLenth              = len(data)                                                                                 
		xlogRecordHeader        = &XLogRecord{}                                                                             
		sizeOfXLogRecord        = uint32(unsafe.Offsetof(xlogRecordHeader.Xl_crc) + unsafe.Sizeof(xlogRecordHeader.Xl_crc)) 
		startPos         uint32 = 0

		sizeOfPageHeader int
		postHalf         uint32
		wals  [256]*singleWal
		count = 0
	)
	if offset < gOffset.offset {
		if (offset + uint64(len(data))) <= gOffset.offset {
			logger.Infof("data should not to decode: offset(%v) + length(%v) < global offset(%v)", offset, uint64(len(data)), gOffset.offset)
			return
		}
		offset = gOffset.offset
	}
	startPos = uint32(offset % uint64(BLOCKSZ))
	if startPos == 0 {

		if uint8(data[0]) == 13 && uint8(data[1]) == 209 {

			if Byte2uint16(data[2:4], nil)&uint16(XLP_LONG_HEADER) == 0 { //short page
				var xlogPageHeaderShort = &XLogPageHeaderShort{}
				var shortSize = unsafe.Sizeof(*xlogPageHeaderShort)
				startPos = Maxalign(uint32(unsafe.Offsetof(xlogPageHeaderShort.XlpRemLen) + unsafe.Sizeof(xlogPageHeaderShort.XlpRemLen)))
				getXlogPageHeaderShort(xlogPageHeaderShort, data[0:startPos])
				if xlogPageHeaderShort.XlpInfo&XLP_FIRST_IS_CONTRECORD != 0 { //the second half of a wal spanning two pages
					if xlogPageHeaderShort.XlpRemLen > uint32(BLOCKSZ-int(shortSize)) {
						PreHalf = append(PreHalf, data[startPos:startPos+uint32(BLOCKSZ-int(shortSize))]...)
						return
					} else {
						postHalf = Maxalign(xlogPageHeaderShort.XlpRemLen)
						PreHalf = append(PreHalf, data[startPos:startPos+postHalf]...)
					}
				}
				totalLenth -= int(startPos)
				sizeOfPageHeader = int(startPos)
			} else { //长页
				var xlogPageHeaderLong = &XLogPageHeaderLong{}
				var longSize = unsafe.Sizeof(*xlogPageHeaderLong)
				startPos = Maxalign(uint32(unsafe.Offsetof(xlogPageHeaderLong.XlpXlogBlcksz) + unsafe.Sizeof(xlogPageHeaderLong.XlpXlogBlcksz)))
				getXlogPageHeaderLong(xlogPageHeaderLong, data[0:startPos])
				if xlogPageHeaderLong.XlpInfo&XLP_FIRST_IS_CONTRECORD != 0 { //the second half of a wal spanning two pages
					if xlogPageHeaderLong.XlpRemLen > uint32(BLOCKSZ-int(longSize)) {
						PreHalf = append(PreHalf, data[startPos:startPos+uint32(BLOCKSZ-int(longSize))]...)
						return
					} else {
						postHalf = Maxalign(xlogPageHeaderLong.XlpRemLen)
						PreHalf = append(PreHalf, data[startPos:startPos+postHalf]...)
					}
				}
				totalLenth -= int(startPos)
				sizeOfPageHeader = int(startPos)

			}

			offset += uint64(startPos)
			if Linked {
				isSpan = true
				logger.Println("parse xlog spanning two pages")
				wals[count] = &singleWal{}
				wg.Add(1)
				decodeRecord(ctx, &PreHalf, sizeOfXLogRecord, PreIno, ino, PreOffset, sizeOfPageHeader, wals[count])
				Linked = false
				count++
				logger.Println("complete the parsing progress")
			}

			startPos += postHalf
			totalLenth -= int(postHalf)
			offset += uint64(postHalf)
			gOffset.offset = offset
		}
	} else {
		totalLenth -= int(startPos)
	}

	for totalLenth >= SIZEOFUINT32 {
		subdata := data[startPos:]
		wals[count] = &singleWal{}
		wg.Add(1)
		go decodeRecord(ctx, &subdata, sizeOfXLogRecord, ino, ino, offset, sizeOfPageHeader, wals[count]) //if a wal is in one page, then firstNode = SecnodNode
		count++
		standedLen := <-walLengthChan
		if standedLen == 0 {
			*quit = true
			break
		}
		if int(standedLen) > totalLenth {
			standedLen = uint32(totalLenth)
		}
		standedLen = Maxalign(standedLen)
		startPos += standedLen
		offset += uint64(standedLen)
		gOffset.offset = offset
		totalLenth -= int(standedLen)

	}
	wg.Wait()

	for i := 0; i < count; i++ {
		if wals[i].pref == "" {
			continue
		}
		modifyRelation(wals[i])
	}
}

func decodeRecord(ctx Context, data *[]byte, sizeOfXLogRecord uint32, firstIno Ino, secondIno Ino, offset uint64, sizeOfPageHeader int, singleWal *singleWal) {
	var (
		rnode        = &RelFileNode{} //block's physical infomation
		block_id     uint8
		startPos     uint32 = 0
		xlogStartPos uint32 = 0

		forkFlags   uint8
		blkBimgInfo uint8
		blkForNum   uint8

		blkHasImage bool

		blkno            uint32
		xlogRecordHeader = &XLogRecord{}
		preIno           Ino
	)
	defer wg.Done()
	startPos = xlogStartPos
	xlogLength := Byte2uint32((*data)[0:4], nil)
	if isSpan {
		isSpan = false
	} else {
		walLengthChan <- xlogLength
	}
	if xlogLength == 0 {
		return
	}
	if len(*data) < int(xlogLength) { //process the first half of a wal spanning two pages	
		PreHalf = make([]byte, len(*data), cap(*data))
		copy(PreHalf, (*data)[:])
		PreIno = firstIno
		Linked = true
		PreOffset = offset
		return
	}

	// get xlogrecordHeader
	xlogRecordtotal := (*data)[startPos : startPos+sizeOfXLogRecord]
	GetXlogRecordHeader(xlogRecordHeader, xlogRecordtotal)

	if firstIno == secondIno {
		secondIno = SAMENODE
	}

	preIno = firstIno

	if xlogRecordHeader.Xl_end <= PreLsn {
		logger.Infof("PreLsn %v, xl_end %v", PreLsn, xlogRecordHeader.Xl_end)
	} else {
		PreLsn = xlogRecordHeader.Xl_end
	}

	startPos += sizeOfXLogRecord

	block_id = (*data)[startPos]

	startPos += 1
	if block_id == XLR_BLOCK_ID_DATA_SHORT {
		//main_data_len, skip
		startPos += 1
	} else if block_id == XLR_BLOCK_ID_DATA_LONG {
		//main_data_len, skip
		Byte2uint32((*data)[startPos:startPos+4], &startPos)
		startPos += 4
	} else if block_id == XLR_BLOCK_ID_ORIGIN {
		//record_origin, skip
		startPos += 2
	} else if block_id == XLR_BLOCK_ID_TOPLEVEL_XID {
		//transactionId, skip
		startPos += 4
	} else if block_id <= XLR_MAX_BLOCK_ID {
		forkFlags = uint8((*data)[startPos])
		blkForNum = forkFlags & BKPBLOCK_FORK_MASK
		blkHasImage = (forkFlags & BKPBLOCK_HAS_IMAGE) != 0
		startPos += 1
		//data_len, skip
		Byte2uint16((*data)[startPos:startPos+2], &startPos)

		if blkHasImage {
			//bimg_len,skip
			Byte2uint16((*data)[startPos:startPos+2], &startPos)
			//hole_offset, skip
			Byte2uint16((*data)[startPos:startPos+2], &startPos)
			//bimg_info, skip
			blkBimgInfo = uint8((*data)[startPos])
			startPos += 1

			if (blkBimgInfo & BKPIMAGE_IS_COMPRESSED) != 0 {
				if (blkBimgInfo & BKPIMAGE_HAS_HOLE) != 0 {
					//hole_length, skip
					Byte2uint16((*data)[startPos:startPos+2], &startPos)
				}
			}
		}
		if (forkFlags & BKPBLOCK_SAME_REL) == 0 {
			rnodeDataLen := unsafe.Sizeof(*rnode)
			rnodeData := (*data)[startPos : uintptr(startPos)+rnodeDataLen]

			getRnode(rnode, rnodeData)
			startPos += uint32(rnodeDataLen)
		}
		blkno = Byte2uint32((*data)[startPos:startPos+4], &startPos)

		pref := strconv.FormatUint(uint64((*rnode).dbNode), 10) + "_" +
			strconv.FormatUint(uint64((*rnode).relNode), 10) + "_" +
			strconv.FormatUint(uint64(blkForNum), 10)

		wal := WalLoc{
			Inode:       preIno,
			Offset:      offset,
			Length:      int(xlogRecordHeader.Xl_tot_len),
			Lsn:         uint64(xlogRecordHeader.Xl_end),
			LinkedInode: secondIno,
		}
		singleWal.pref = pref
		singleWal.blkno = blkno
		singleWal.wal = wal

	}
}

func modifyRelation(sw *singleWal) {

	pref := sw.pref
	blkno := sw.blkno
	key := pref + "_" + strconv.FormatUint(uint64(blkno), 10)
	wal := sw.wal
	logger.Infoln("sw is ", sw, sw.pref, sw.wal, sw.blkno, key)
	pages := PagesWithWal.get(pref)
	var existWal bool
	if pages != nil {
		if !pages.get(int(blkno)) {
			pages.set(int(blkno))
		} else {
			existWal = true
		}

	} else {
		v := &Set{s: map[int]Empty{int(blkno): empty}}
		PagesWithWal.set(pref, v)
	}
	if _, ok := NeedFlushPages[NeedFlushIdx][key]; !ok {
		NeedFlushPages[NeedFlushIdx][key] = empty
	}
	if p, ok := Pages.Get(key); ok {
		page := p.(*PageInfo)
		page.wals = append(page.wals, wal)
		page.lastVisit = time.Now().Unix()
		page.visits += 1
	} else {
		var newWals []WalLoc
		var wals []WalLoc
		if existWal {
			vals, err := rdbClient.LRange(context.Background(), key, 0, -1).Result()
			if err != nil {
				if err == redis.Nil {
					logger.Infoln("empty links in redis when write wal record.")
				} else {
					logger.Errorf("query redis failed when page is not in memory: %v", err)
					return
				}

			}

			for _, v := range vals {
				var w WalLoc
				err = json.Unmarshal([]byte(v), &w)
				if err != nil {
					logger.Errorf("Unmarshal wal failed: %s", err)
					return
				}
				wals = append(wals, w)
			}
			newWals = append(wals, wal)
		} else {
			newWals = append(newWals, wal)
		}

		newPage := &PageInfo{
			wals:      newWals,
			lastVisit: time.Now().Unix(),
			visits:    1,
			flushed:   len(wals),
		}
		Pages.Set(key, newPage)
	}

}

func GetXlogRecordHeader(xlogRecHeader *XLogRecord, bytes []byte) {
	var start uint32 = 0
	xlogRecHeader.Xl_tot_len = Byte2uint32(bytes[0:4], &start)
	xlogRecHeader.Xl_xid = TransactionId(Byte2uint32(bytes[start:start+4], &start))
	xlogRecHeader.Xl_prev = XLogRecPtr(Byte2uint64(bytes[start:start+8], &start))
	xlogRecHeader.Xl_end = XLogRecPtr(Byte2uint64(bytes[start:start+8], &start))

	xlogRecHeader.Xl_info = uint8(bytes[start])
	start++
	if uint8(bytes[start]) > 0 {
		xlogRecHeader.mtr = true
	} else {
		xlogRecHeader.mtr = false
	}
	start++
	xlogRecHeader.blockNum = uint8(bytes[start])
	start++
	xlogRecHeader.Xl_rmid = RmgrId(uint8(bytes[start]))
	start++
	xlogRecHeader.Xl_crc = pg_crc32c(Byte2uint32(bytes[start:start+4], &start))
}

func getXlogPageHeaderShort(short *XLogPageHeaderShort, bytes []byte) {
	var start uint32 = 0
	short.XlpMagic = Byte2uint16(bytes[0:2], &start)
	short.XlpInfo = Byte2uint16(bytes[start:start+2], &start)
	short.XlpTli = Byte2uint32(bytes[start:start+4], &start)
	short.XlpPageAddr = Byte2uint64(bytes[start:start+8], &start)
	short.XlpRemLen = Byte2uint32(bytes[start:start+4], &start)
}
func getXlogPageHeaderLong(long *XLogPageHeaderLong, bytes []byte) {
	var start uint32 = 0
	long.XlpMagic = Byte2uint16(bytes[0:2], &start)
	long.XlpInfo = Byte2uint16(bytes[start:start+2], &start)
	long.XlpTli = Byte2uint32(bytes[start:start+4], &start)
	long.XlpPageAddr = Byte2uint64(bytes[start:start+8], &start)
	long.XlpRemLen = Byte2uint32(bytes[start:start+4], &start)
	long.XlpSysId = Byte2uint64(bytes[start:start+8], &start)
	long.XlpSegSize = Byte2uint32(bytes[start:start+4], &start)
	long.XlpXlogBlcksz = Byte2uint32(bytes[start:start+4], &start)

}

func getRnode(node *RelFileNode, bytes []byte) {
	var start uint32 = 0
	node.spcNode = Oid(Byte2uint32(bytes[start:start+4], &start))
	node.dbNode = Oid(Byte2uint32(bytes[start:start+4], &start))
	node.relNode = Oid(Byte2uint32(bytes[start:start+4], &start))
}

func Maxalign(size uint32) uint32 {
	return (size + 7) & ^uint32(7)
}
func Byte2uint16(data []byte, startPos *uint32) uint16 {
	if startPos != nil {
		*startPos += 2
	}
	return uint16(data[0]) | uint16(data[1])<<8
}

func Byte2uint32(data []byte, startPos *uint32) uint32 {
	if startPos != nil {
		*startPos += 4
	}
	return uint32(data[0]) | uint32(data[1])<<8 | uint32(data[2])<<16 | uint32(data[3])<<24
}

func Byte2uint64(data []byte, startPos *uint32) uint64 {
	if startPos != nil {
		*startPos += 8
	}
	return uint64(data[0]) | uint64(data[1])<<8 | uint64(data[2])<<16 | uint64(data[3])<<24 |
		uint64(data[4])<<32 | uint64(data[5])<<40 | uint64(data[6])<<48 | uint64(data[7])<<56
}
